package com.study.flink.java.day06_exactly;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;
import java.util.Iterator;

/**
 * 只实现数据一次的处理的自定义并行的source
 */
public class MyParFileExactlyOnceSource extends RichParallelSourceFunction<Tuple2<String, String>> implements CheckpointedFunction {

    private String path;
    private Boolean flag = true;
    private Long offset = 0L;
    // 状态数据不参与序列化必须加 transient 关键字
    private transient ListState<Long> offsetState;

    public MyParFileExactlyOnceSource() {
    }

    public MyParFileExactlyOnceSource(String path) {
        this.path = path;
    }

    // 就执行一次
    @Override
    public void run(SourceContext<Tuple2<String, String>> sourceContext) throws Exception {
        // 获取offsetState历史值
        Iterator<Long> it = offsetState.get().iterator();
        while (it.hasNext()) {
            offset = it.next();
        }

        int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
        // dir/day06/mysource
        RandomAccessFile randomAccessFile = new RandomAccessFile(path + "/" + subtaskIndex + ".txt", "r");

        // 从指定的位置读取数据
        randomAccessFile.seek(offset);

        // 获取一个锁，多线程更新偏移量需要加锁
        final Object lock = sourceContext.getCheckpointLock();

        while (flag) {
            String line = randomAccessFile.readLine();
            if (line != null) {
                line = new String(line.getBytes("ISO-8859-1"), "UTF-8");
                synchronized (lock) {
                    // 文件指针赋值给offset
                    offset = randomAccessFile.getFilePointer();
                    // 将数据发送出去
                    sourceContext.collect(Tuple2.of(subtaskIndex + "", line));
                }
            } else {
                Thread.sleep(1000L);
            }
        }

    }

    @Override
    public void cancel() {
        flag = false;
    }

    /**
     * 由JobManager触发
     * 定期将指定的状态数据保存到StateBackEnd中
     *
     * @param functionSnapshotContext
     * @throws Exception
     */
    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        // 1.将历史值清除
        offsetState.clear();

        // 2.更新最新的状态值
        // offsetState.update(Collections.singletonList(offset));
        offsetState.add(offset);
    }

    /**
     * 初始化OperatorState，生命周期方法，构造方法执行后会执行一次
     *
     * @param functionInitializationContext
     * @throws Exception
     */
    @Override
    public void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {

        // 定义一个状态描述器(偏移量)
        ListStateDescriptor<Long> stateDescriptor = new ListStateDescriptor<Long>(
                "offset-state",
                //TypeInformation.of(new TypeHint<Long>() {})
                //Long.class
                Types.LONG
        );

        // 初始化状态或获取历史状态(OperatorState)
        offsetState = functionInitializationContext.getOperatorStateStore().getListState(stateDescriptor);

    }


}
